package com.ustc.hewei.detectbeats.detect;

import com.alibaba.fastjson2.JSON;
import com.ustc.hewei.rpc.client.net.SendHandler;
import com.ustc.hewei.rpc.common.Utils.MessageSerialize;
import com.ustc.hewei.rpc.common.protocol.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

/**
* @description: 利用缓存发送心跳
* @author hewei
* @date 2022/8/17 20:25
* @version 1.0
*/

@Slf4j
public class SendBeat implements Runnable{
    @Resource
    StringRedisTemplate stringRedisTemplate;

    // 存放所有的服务列表，键为IP，值为ip对应的所有的服务在redis中的键
    private final ConcurrentHashMap<String, List<String>> serviceMap;

    private final String protocol;

    // 失效的服务的计数，键为ip，值为连续失效的次数
    private final HashMap<String, Integer> invalidIpCount;

    // 最大失效次数
    private final int MAX_COUNT;

    // 存放待删除的服务ip的阻塞队列，开一个单线程任务来删除
    private final BlockingQueue<String> removeIpList;

    public SendBeat(ConcurrentHashMap<String, List<String>> serviceMap, String protocol, int MAX_COUNT) {
        this.serviceMap = serviceMap;
        this.protocol = protocol;
        invalidIpCount = new HashMap<>();
        this.MAX_COUNT = MAX_COUNT;
        removeIpList = new ArrayBlockingQueue<>(1024 * 1024);
        Executors.newFixedThreadPool(1).submit(this);
    }

    /*
     * @description: 发送心跳信号并对响应进行判断
     * @author: hewei
     * @date: 2022/8/19 21:19
     **/
    public void send() {
        if (serviceMap.isEmpty()) {
            log.info("注册中心暂无服务！");
            return;
        }
        for (Map.Entry<String, List<String>> entry : serviceMap.entrySet()) {
            String ip = entry.getKey();
            List<String> keys = entry.getValue();
            RequestWarp requestWarp = new RequestWarp();
            requestWarp.setRequestType(RequestType.KEEP_ALIVE);
            byte[] data = MessageSerialize.objectToByteArray(requestWarp);
            try {
                byte[] response = sendRequest(data, ip);
                ResponseWarp responseWarp = MessageSerialize.byteArrayToResp(response);
                if (responseWarp.getStatus() != Status.KEEP_ALIVE) {
                    log.error("心跳检测异常！");
                    removedService(keys, ip);
                    log.info("第 {} 次心跳检测{}失败", invalidIpCount.get(ip), ip);
                } else if (invalidIpCount.containsKey(ip)){
                    log.info("服务地址 {} ,重连成功！", ip);
                    addService(keys, ip);
                    log.info(" {} 的服务已重新写入！", ip);
                    invalidIpCount.remove(ip);
                }
            } catch (Exception e) {
                log.error("心跳检测异常！");
                removedService(keys, ip);
                log.info("第{}次心跳检测 {} 失败", invalidIpCount.get(ip), ip);
            }
        }
    }

    /*
     * @description: 在redis中删除ip所对应的所有服务并且将服务的失效次数加一
     * @param keys 所有服务对应的redis中的键
     * @param ip 要删除的ip
     * @author: hewei
     * @date: 2022/8/18 22:01
     **/
    public void removedService(List<String> keys, String ip) {
        if (!invalidIpCount.containsKey(ip)) {
            HashMap<String, String> map = new HashMap<>();
            for (String key : keys) {
                Service service = new Service();
                // 将redis中的键进行分割，得到服务的名称
                String serviceName = key.substring(key.indexOf("/") + 1, key.lastIndexOf("/"));
                service.setAddress(ip).setName(serviceName).setProtocol(protocol);
                String jsonString = JSON.toJSONString(service);
                map.put(key, jsonString);
            }
            for (Map.Entry<String, String> removeEntry : map.entrySet()) {
                stringRedisTemplate.opsForSet().remove(removeEntry.getKey(), removeEntry.getValue());
            }
        }
        // 如果当前失效次数已经达到最大，则会将其放入到阻塞队列中进行删除操作
        invalidIpCount.put(ip, invalidIpCount.getOrDefault(ip, 0) + 1);
        if (invalidIpCount.get(ip) >= MAX_COUNT) {
            removeIpList.offer(ip);
        }
    }

    /*
     * @description: 当已经断连的服务在最大超时时间之内重新连接，则将该ip下的所有服务重新写入到redis中
     * @param keys ip对应的服务的所有的在redis中的键
     * @param ip 恢复过来的ip
     * @author: hewei
     * @date: 2022/8/19 21:15
     **/
    public void addService(List<String> keys, String ip) {
        for (String key : keys) {
            Service service = new Service();
            // 将redis中的键进行分割，得到服务的名称
            String serviceName = key.substring(key.indexOf("/") + 1, key.lastIndexOf("/"));
            service.setAddress(ip).setName(serviceName).setProtocol(protocol);
            String jsonString = JSON.toJSONString(service);
            stringRedisTemplate.opsForSet().add(key, jsonString);
        }
    }

    /*
     * @description: 发送心跳信号
     * @param data 心跳包装类的序列化数组
     * @param fullIp 服务端的ip，即需要发送的地址
     * @return byte[] 服务端返回的响应
     * @author: hewei
     * @date: 2022/8/19 21:17
     **/
    public byte[] sendRequest(byte[] data, String fullIp) {
        SendHandler sendHandler = new SendHandler(data);
        byte[] response;

        //将地址进行分割，因为形式是"ip：port"
        String[] split = fullIp.split(":");
        int port = Integer.parseInt(split[1]);
        String ip = split[0];

        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            new Bootstrap().group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    //设置可以接收的大小，默认为2048个字节，不够用
                    //.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(65535))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //调用自定义的消息处理器
                            socketChannel.pipeline().addLast(sendHandler);
                        }
                    }).connect(ip, port).sync();
            response = (byte[]) sendHandler.getResponse();
            log.info("从 {} 获取心跳成功！", fullIp);
        } catch (InterruptedException e) {
            throw new RuntimeException("心跳检测网络连接异常！", e);
        } finally {
            group.shutdownGracefully();
        }
        return response;
    }

    @Override
    public void run() {
        while (true) {
            try {
                String ip = removeIpList.take();
                serviceMap.remove(ip);
                invalidIpCount.remove(ip);
                log.info("{} 下的服务已从注册中心删除！", ip);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public HashMap<String, Integer> getInvalidIpCount() {
        return invalidIpCount;
    }
}
